package cn.hty.mymsg.handler.flowcontrol;

import cn.hty.mymsg.common.constant.CommonConstant;
import cn.hty.mymsg.common.domain.TaskInfo;
import cn.hty.mymsg.common.enums.ChannelType;
import cn.hty.mymsg.common.enums.EnumUtil;
import cn.hty.mymsg.handler.enums.RateLimitStrategy;
import cn.hty.mymsg.handler.flowcontrol.annotations.LocalRateLimit;
import cn.hty.mymsg.service.ConfigService;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.util.concurrent.RateLimiter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.aop.support.AopUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
@Service
@Slf4j
public class FlowControlFactory implements ApplicationContextAware {
    private static final String FLOW_CONTROL_KEY = "flowControlRule";
    private static final String FLOW_CONTROL_PREFIX = "flow_control_";
    private ApplicationContext applicationContext;
    private final Map<RateLimitStrategy, FlowControlService> flowControlServiceMap = new ConcurrentHashMap<>();

    @Autowired
    private ConfigService config;
    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }

    /**
     * 判断限流值和默认是否一样，不一样按照配置的来
     * @param taskInfo
     * @param flowControlParam
     */
    public void flowControl(TaskInfo taskInfo, FlowControlParam flowControlParam){
        Double rateInitValue = flowControlParam.getRateInitValue();
        Double rateLimitConfig = getRateLimitConfig(taskInfo.getSendChannel());
        if (Objects.nonNull(rateLimitConfig)&&!rateInitValue.equals(rateLimitConfig)){
            RateLimiter  rateLimiter = RateLimiter.create(rateLimitConfig);
            flowControlParam.setRateLimiter(rateLimiter);
            flowControlParam.setRateInitValue(rateLimitConfig);
        }
        FlowControlService flowControlService = flowControlServiceMap.get(flowControlParam.getRateLimitStrategy());
        if (Objects.isNull(flowControlService)) {
            log.error("没有找到对应的单机限流策略");
            return;
        }
        double costTime = flowControlService.flowControl(taskInfo, flowControlParam);
        if (costTime > 0) {
            log.info("consumer {} flow control time {}",
                    EnumUtil.getEnumByCode(taskInfo.getSendChannel(), ChannelType.class).getDescription(), costTime);
        }
    }

    /**
     * 获取配置限流值
     * @param channelCode
     * @return
     */
    private Double getRateLimitConfig(Integer channelCode){
        String flowControlConfig  = this.config.getProperty(FLOW_CONTROL_KEY, CommonConstant.EMPTY_JSON_OBJECT);
        JSONObject jsonObject = JSON.parseObject(flowControlConfig);
        if (Objects.isNull(jsonObject.getDouble(FLOW_CONTROL_PREFIX + channelCode))){
            return null;
        }
        return jsonObject.getDouble(FLOW_CONTROL_PREFIX + channelCode);

    }
    /**
     * 初始化扫描并注册带有特定注解的Bean
     */
    @PostConstruct
    private void init(){
        //获取bean
        Map<String, Object> serviceMap = this.applicationContext.getBeansWithAnnotation(LocalRateLimit.class);
        //遍历bean，检查是否都继承了FlowControlService，并处理
        serviceMap.forEach((name,service)->{
            if (service instanceof FlowControlService){
                //获取 LocalRateLimit 注解并提取限流策略
                LocalRateLimit localRateLimit = AopUtils.getTargetClass(service).getAnnotation(LocalRateLimit.class);
                //限流策略
                RateLimitStrategy rateLimitStrategy = localRateLimit.rateLimitStrategy();
                flowControlServiceMap.put(rateLimitStrategy, (FlowControlService) service);
            }
        });

    }
}
